package com.zhf.config;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Component
public class KafkaConfig {

    @Value("${my.kafka.topic}")
    private String kafkaTopic;

    @Value("${my.kafka.address}")
    private String kafkaAddress;

    @Value("${my.kafka.batch.size}")
    private int kafkaBatchSize;

    @Value("${my.kafka.key.serializer}")
    private String kafkaKeySerializer;

    @Value("${my.kafka.value.serializer}")
    private String kafkaValueSerializer;

    @Value("${my.kafka.retry.times}")
    private String kafkaRetryTimes;

    @Value("${my.kafka.acks}")
    private String kafkaAcks;

    @Value("${my.kafka.buffer.memory}")
    private String kafkaMemory;

    public String getKafkaTopic() {
        return kafkaTopic;
    }

    public void setKafkaTopic(String kafkaTopic) {
        this.kafkaTopic = kafkaTopic;
    }

    public String getKafkaAddress() {
        return kafkaAddress;
    }

    public void setKafkaAddress(String kafkaAddress) {
        this.kafkaAddress = kafkaAddress;
    }

    public int getKafkaBatchSize() {
        return kafkaBatchSize;
    }

    public void setKafkaBatchSize(int kafkaBatchSize) {
        this.kafkaBatchSize = kafkaBatchSize;
    }

    public String getKafkaKeySerializer() {
        return kafkaKeySerializer;
    }

    public void setKafkaKeySerializer(String kafkaKeySerializer) {
        this.kafkaKeySerializer = kafkaKeySerializer;
    }

    public String getKafkaValueSerializer() {
        return kafkaValueSerializer;
    }

    public void setKafkaValueSerializer(String kafkaValueSerializer) {
        this.kafkaValueSerializer = kafkaValueSerializer;
    }

    public String getKafkaRetryTimes() {
        return kafkaRetryTimes;
    }

    public void setKafkaRetryTimes(String kafkaRetryTimes) {
        this.kafkaRetryTimes = kafkaRetryTimes;
    }

    public String getKafkaAcks() {
        return kafkaAcks;
    }

    public void setKafkaAcks(String kafkaAcks) {
        this.kafkaAcks = kafkaAcks;
    }

    public String getKafkaMemory() {
        return kafkaMemory;
    }

    public void setKafkaMemory(String kafkaMemory) {
        this.kafkaMemory = kafkaMemory;
    }

    public Properties initKafkaProperties(){
        Properties properties = new Properties();
        properties.put("bootstrap.servers", kafkaAddress);
        properties.put("retries", kafkaRetryTimes);
        properties.put("batch.size", kafkaBatchSize);
        properties.put("acks", kafkaAcks);
        properties.put("buffer.memory", kafkaMemory);
        properties.put("key.serializer", kafkaKeySerializer);
        properties.put("value.serializer", kafkaValueSerializer);
        return properties;
    }
}
